package com.heima.article.service.impl;


import com.heima.article.service.ApCollectionBehaviorService;
import com.heima.common.contants.MQConstants;
import com.heima.common.contants.RedisConstants;
import com.heima.common.dtos.AppHttpCodeEnum;
import com.heima.common.dtos.ResponseResult;
import com.heima.common.exception.LeadNewsException;
import com.heima.model.article.dtos.UpdateArticleMsg;
import com.heima.model.behavior.dtos.CollectionBehaviorDto;
import com.heima.model.user.pojos.ApUser;
import com.heima.utils.common.JsonUtils;
import com.heima.utils.common.ThreadLocalUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class ApCollectionBehaviorServiceImpl implements ApCollectionBehaviorService {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Autowired
    private StringRedisTemplate redisTemplate;

    //文章收藏
    @Override
    public ResponseResult collectionBehavior(CollectionBehaviorDto dto) {
        try {
            //判断是否登录
            ApUser user = (ApUser) ThreadLocalUtils.get();
            if (user == null){
                throw new LeadNewsException(AppHttpCodeEnum.NEED_LOGIN);
            }
            //校验参数
            if (dto.getEntryId() == null){
                throw new LeadNewsException(AppHttpCodeEnum.PARAM_REQUIRE);
            }

            //存入redis
            String key = RedisConstants.ARTICLE_COLLECTION+dto.getEntryId();
            Object o = redisTemplate.opsForHash().get(key, user.getId().toString());

            if (dto.getOperation()==0){
                //收藏
                if (o != null){
                    return ResponseResult.okResult("你已经收藏，无需再收藏");
                }
                //写入redis 和发送消息
                String value = dto.getEntryId().toString()+user.getId();
               redisTemplate.opsForHash().put(key,user.getId().toString(),value);

                //发送消息给KafkaStream进行数据同步
                UpdateArticleMsg articleMsg = new UpdateArticleMsg();
                articleMsg.setArticleId(dto.getEntryId());
                articleMsg.setType(UpdateArticleMsg.UpdateArticleType.COLLECTION);
                kafkaTemplate.send(MQConstants.HOT_ARTICLE_INPUT_TOPIC, JsonUtils.toString(articleMsg));
            }else {
                //取消收藏
                if (o == null){
                    return ResponseResult.okResult("你没有收藏，无需取消收藏");
                }

                redisTemplate.opsForHash().delete(key,user.getId().toString());
                //TODO 发送Kafka

            }
            return ResponseResult.okResult(null);

        } finally {
            ThreadLocalUtils.remove();
        }
    }
}
